package com.rem.kafka.listener;

import com.rem.kafka.controller.User;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;

/**
 * @author Rem
 * @date 2023-01-06
 */
@Configuration
public class KafkaListen {

    /**
     * 代码中的配置优先级高于配置文件的配置
     * 自动提交offset,并且批最多量拉50个
     * spring.kafka.listener.type=batch
     * spring.kafka.consumer.enable-auto-commit=true
     * spring.kafka.consumer.max-poll-records=50
     * <p>
     * 同时监听first,four 两个topic,也可以通过
     * topicPartitions = {
     *
     * @param records
     * @TopicPartition(topic = "three", partitions = {"1"}),
     * @TopicPartition(topic = "four", partitions = {"1", "2"}, partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "20")
     * 来进行配置topicPartitions和topics 不能共存
     */
    @KafkaListener(topics = {"first", "four"})
    public void listen1(List<ConsumerRecord<String, User>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, User> record = records.get(i);
            User value = record.value();
            System.out.println(i + ":" + record);
            System.out.println(i + ":" + value);
        }
    }

    /**
     * 手动提交offset
     *
     * @param record
     * @param ack
     */
    public void listen2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手动提交offset
        ack.acknowledge();
    }
}
